package streamforce.crypto.cloud;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.crypto.KPAbe;
import streamforce.crypto.cloud.subscriber.CloudJoinPriceStage1;
import streamforce.crypto.cloud.subscriber.CloudJoinPriceStage2;
import streamforce.crypto.cloud.subscriber.UserCiphertextSubscriber;
import streamforce.crypto.cloud.subscriber.UserInitSubscriber;
import streamforce.crypto.message.PolicyMessage;
import streamforce.crypto.message.UserHandshakeMessage;

public class EPCloudServer {
	private static final Log log = LogFactory.getLog(EPCloudServer.class);
	private int port;
    private int threadCore;
    private int queueMax;   

    public static final String CONFIG_FILE = "config.xml";
    public static boolean QUEUE_DUMP;
    public static final int DEFAULT_PORT = 6789;
    public static final int NCORES = Runtime.getRuntime().availableProcessors();
    public static final Properties MODES = new Properties();
    
    //store the entry time of each tuple
    public static final Map<Tuple, TimeTuple> tupleTime = new HashMap<Tuple, TimeTuple>();     
    public static String queueStatFile=""; 
    
    private ThreadPoolExecutor executor;//can be null

    private EPCloud cloudInstance;
    
    //connection to the data user
    private Map<Integer, ObjectOutputStream> userConnectionsMap; 

    public static ArrayList subscribers; 
    KPAbe cryptoHandler; 
    Stat stat; 
    
    public EPCloudServer(String file) throws Exception{
    	Properties prop = new Properties();
    	prop.load(new FileInputStream(file)); 
    	this.port = new Integer(prop.getProperty("serverPort")).intValue(); 
    	QUEUE_DUMP = new Boolean(prop.getProperty("queueStat")).booleanValue(); 
    	this.threadCore = 5;
    	this.queueMax = 10;
    	
    	Timer t = new Timer("Server tasks", true);    	
    	t.scheduleAtFixedRate(new TimerTask(){
    		public void run(){
    			Stat.aggregateAndResetAll("nEventReceived"); 
    			Stat.aggregateAndResetAll("nTransformed"); 
    			Stat.aggregateAndResetAll("nTransformed_join");
    			Stat.aggregateAndResetAll("nJoins"); 
    			Stat.aggregateAndResetAll("nACInstances");
    			Stat.aggregateAndResetAll("nTuples"); 
    		}
    	}, 0, 1000);
    	
    	cloudInstance = new EPCloud();
    	cloudInstance.initializeEPServiceProvider(); 
    	
    	subscribers = new ArrayList();  
    	this.userConnectionsMap = new HashMap<Integer, ObjectOutputStream>();    	 
    	
    	this.cryptoHandler = new KPAbe();
    	stat = new Stat("nACInstances"); 
    }
    
    //starts receiving connection from the client and fires new thread to handle it
    public void start(){
    	this.executor = new ThreadPoolExecutor(this.threadCore, this.threadCore*NCORES, 
    			0, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(queueMax), 
    			new ThreadFactory() {
    				long count = 0;
    				public Thread newThread(Runnable r) {    					
    					return new Thread(r, "EsperServer-" + count++);
    				}
        		},
        		new ThreadPoolExecutor.CallerRunsPolicy() {
        			public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        				super.rejectedExecution(r, e);
        			}
        		});
    	executor.prestartAllCoreThreads();
    	final int basePort = this.port;
    	new Thread(new Runnable(){
			public void run() {
				startServerForOwner(basePort);
			}
    	}).start();
    	
    	new Thread(new Runnable(){
			public void run() {
				startServerForUser(basePort+1);
			}
    	}).start();
    }
    
    private void startServerForOwner(int port){    	
		try {
			ServerSocket ss = new ServerSocket(port);
			
			do{
				Socket clientSocket = ss.accept(); 				
				log.info("Owner connection established");
				new EPOwnerConnection(clientSocket, this.cloudInstance, this.executor,this.cryptoHandler).start();
								
			}while (true); 
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}	
    }
    
    private void startServerForUser(int port){    	
		try {
			ServerSocket ss = new ServerSocket(port);
			
			do{
				Socket clientSocket = ss.accept(); 		
				log.info("User connection established"); 
				ObjectOutputStream oos = new ObjectOutputStream(clientSocket.getOutputStream());
				ObjectInputStream ois = new ObjectInputStream(clientSocket.getInputStream());					
				Object o = ois.readObject(); 				
				if (o instanceof UserHandshakeMessage){
					UserHandshakeMessage uhsm = (UserHandshakeMessage)o; 
					this.initializeUser(uhsm, oos);
					if (uhsm.getType() == PolicyMessage.JOIN_QUERY){
						Object o1 = ois.readObject(); 
						if (o1 instanceof PolicyMessage){
							this.initializeUserJoin((PolicyMessage)o1, oos);
						}
					}
				}												
			}while (true); 
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}	
    }
    
    //register UserInitListener and UserCiphertextListener
    private void initializeUser(UserHandshakeMessage uim, ObjectOutputStream oos){
    	int uid = uim.getUserIds()[0];    	
    	UserInitSubscriber uis = new UserInitSubscriber(oos, this.executor);
    	String exp = "", exp1 = "";
    	switch (uim.getType()){
    	case PolicyMessage.SIMPLE_QUERY:
    		exp = "select * from UserInitEvent(userId="+uid+")";
    		exp1 = "select * from TransformedCTEventAll(userId="+uid+")";
    		break;
    	case PolicyMessage.AGG_QUERY_1:
    		exp = "select * from UserInitAggEvent(userId="+uid+")";
    		exp1 = "select * from TransformedCTEventAll(userId="+uid+")";
    		break;
    	case PolicyMessage.AGG_QUERY_2://same as simple queries
    		exp = "select * from UserInitEvent(userId="+uid+")";
    		exp1 = "select * from TransformedCTEventAll(userId="+uid+")";    		
    		break;
    	case PolicyMessage.AGG_QUERY_3:
    		exp = "select * from UserInitEvent(userId="+uid+")";
    		exp1 = "select * from TransformedCTEventAllAgg3(userId="+uid+")";
    		break;
    	}   
    	
    	if (uim.getType() == PolicyMessage.JOIN_QUERY){
    		exp = "select * from UserInitEvent(userId="+uim.getUserIds()[0]+")";
    		exp1 = "select * from UserInitEvent(userId="+uim.getUserIds()[1]+")";
    		this.cloudInstance.registerQueryPattern(exp, uis);
    		this.cloudInstance.registerQueryPattern(exp1, new UserInitSubscriber(oos, this.executor));
    		    		
			String exp2 = "select * from TransformedCTEventJoin(userId=" + uid
					+ " or userId=" + uim.getUserIds()[1] + ")";
    		this.cloudInstance.registerQueryPattern(exp2, new UserCiphertextSubscriber(oos, this.executor));
    	}
    	else {
			this.cloudInstance.registerQueryPattern(exp, uis);
			UserCiphertextSubscriber ucs = new UserCiphertextSubscriber(oos,
					this.executor);
			this.cloudInstance.registerQueryPattern(exp1, ucs);
		}
    	log.info("user connection "+uid+ " established");
    }
   
    //this is where we intialize AC instance, it having been deferred when the cloud
    //received PolicyMessage from the owner
    private void initializeUserJoin(PolicyMessage pm, ObjectOutputStream oos){    	
    	String exp1 = "select * from StockEvent(stockId="+pm.getStreamIds()[0]+")";
		String exp2 = "select * from StockEvent(stockId="+pm.getStreamIds()[1]+")";
		
		long start = System.currentTimeMillis(); 
		CloudJoinPriceStage1 jppc = new CloudJoinPriceStage1(pm.getUserIds()[0], this.cryptoHandler);
		jppc.initACInstance(pm.getStreamIds()[0], pm.getUserIds()[0],  pm.getPubK()[0], pm.gettK()[0], pm.getDetK()[0]);
		this.cloudInstance.registerQueryPattern(exp1, jppc);
		this.stat.update(); 
		
		CloudJoinPriceStage1 jppc1 = new CloudJoinPriceStage1(pm.getUserIds()[1], this.cryptoHandler);
		jppc.initACInstance(pm.getStreamIds()[1], pm.getUserIds()[1], pm.getPubK()[1], pm.gettK()[1], pm.getDetK()[1]);
		
		long end = System.currentTimeMillis(); 
		log.info("init join query time = "+(end-start)); 
		
		this.cloudInstance.registerQueryPattern(exp2, jppc1);
		this.stat.update();
		
		/*String expj = "select s1.timestamp, s1.stockId, s2.timestamp, s2.stockId, s1.price('ws=1'), s1.volume('ws=1'), s2.volume('ws=1')"
				+ " from StockEventJoin(stockId="
				+ pm.getStreamIds()[0]
				+ ").win:length(5) as s1, StockEventJoin(stockId="
				+ pm.getStreamIds()[1]
				+ ").win:length(5) as s2"
				+ " where s1.price('price_det') = s2.price('price_det')";*/		
		CloudJoinPriceStage2 jppcj = new CloudJoinPriceStage2(pm.getUserIds()[0], pm.getUserIds()[1], this.cryptoHandler);						
		this.cloudInstance.registerQueryPattern(pm.getQueryExpression(), jppcj);
		
    }
        
    
    public static void main(String[] args) throws Exception{
    	(new EPCloudServer(args[0])).start(); 
    }
}
